// =====================================================================================
//	Copyright (C) 2019 by Jiaxing Shao.All rights reserved.
//	文 件 名:  websocket_server.cpp
//	作    者:  Jiaxing Shao, 13315567369@163.com
//	版 本 号:  1.0
//	创建时间:  2019年05月06日 18时14分37秒
//	Compiler:  g++
//	描    述:
// =====================================================================================

#include "websocket_server.h"

WebSocketServer::WebSocketServer()
{
    m_server.init_asio();
    m_server.set_reuse_addr(true);
    m_server.set_open_handler(bind(&WebSocketServer::on_open, this, ::_1));
    m_server.set_fail_handler(bind(&WebSocketServer::on_fail, this, ::_1));
    m_server.set_close_handler(bind(&WebSocketServer::on_close, this, ::_1));
    m_server.set_message_handler(bind(&WebSocketServer::on_message, this, ::_1, ::_2));
}

WebSocketServer::~WebSocketServer()
{
}

void WebSocketServer::on_open(connection_hdl hdl)
{
    {
        lock_guard<mutex> guard(m_action_lock);
        m_actions.push(action(SUBSCRIBE, hdl));
    }
    m_action_cond.notify_one();
}

void WebSocketServer::on_fail(connection_hdl hdl)
{
    printf("in  fail\n");
    printf("out fail\n");
}

void WebSocketServer::on_close(connection_hdl hdl)
{
    {
        lock_guard<mutex> guard(m_action_lock);
        m_actions.push(action(UNSUBSCRIBE, hdl));
    }
    m_action_cond.notify_one();
}

void WebSocketServer::on_message(connection_hdl hdl, message_ptr msg)
{
    {
        lock_guard<mutex> guard(m_action_lock);
        m_actions.push(action(MESSAGE, hdl, msg));
    }
    m_action_cond.notify_one();
}

void WebSocketServer::start(int port)
{
    thread t(process_message, this);
    m_server.listen(port);
    m_server.start_accept();
    try {
        m_server.run();
    } catch (const std::exception &e) {
        std::cout << e.what() << endl;
    }
    t.join();
}

void WebSocketServer::send_message(char *pData, int nSize)
{
    lock_guard<mutex> guard(m_connection_lock);

    conn_list::iterator iter;
    for (iter = m_connections.begin(); iter != m_connections.end();) {
        try {
            m_server.send(*iter, pData, nSize, websocketpp::frame::opcode::binary);
            iter++;
        } catch (const std::exception &e) {
            iter = m_connections.erase(iter);
            printf("send_message error.\n");
        }
    }
}

void *WebSocketServer::process_message(void *arg)
{
    WebSocketServer *pthis = (WebSocketServer *)arg;
    while (true) {
        unique_lock<mutex> lock(pthis->m_action_lock);
        while (pthis->m_actions.empty()) {
            pthis->m_action_cond.wait(lock);
        }
        action act = pthis->m_actions.front();
        pthis->m_actions.pop();
        lock.unlock();

        if (act.type == SUBSCRIBE) {
            lock_guard<mutex> guard(pthis->m_connection_lock);
            pthis->m_connections.insert(act.hdl);
        } else if (act.type == UNSUBSCRIBE) {
            lock_guard<mutex> guard(pthis->m_connection_lock);
            pthis->m_connections.erase(act.hdl);
        } else if (act.type == MESSAGE) {
            lock_guard<mutex>   guard(pthis->m_connection_lock);
            conn_list::iterator iter;
            for (iter = pthis->m_connections.begin(); iter != pthis->m_connections.end(); iter++) {
                pthis->m_server.send(*iter, act.msg);
            }
        } else {
            // undefined
        }
    }
}
